package com.atguigu.chapter02;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/8 10:21
 */
public class Flink04_Stream_UnBounded_Lambda_WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 各种转换
        DataStreamSource<String> lineStream = env.socketTextStream("hadoop162", 9999);
        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = lineStream
            .flatMap((String line, Collector<String> collector) -> {
                for (String word : line.split(" ")) {
                    collector.collect(word);
                }
            }).returns(Types.STRING)
            // 由于返回值类型Tuple中有泛型,在运行的时候会被擦除, 导致flink无法推导出他的类型, 保存
            // 解决:1. 使用匿名内部类 2. 使用外部类(代码比较多)  3. 显示的指定泛型的类型 调用 returns
            .map(word -> Tuple2.of(word, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG))
            //            .map(new MyMapFunction())
            .keyBy(t -> t.f0)
            .sum(1);
        
        // 3. 输出
        resultStream.print();
        
        // 4. 启动执行环境
        env.execute();
    }
    
    public static class MyMapFunction implements MapFunction<String, Tuple2<String, Long>> {
        
        @Override
        public Tuple2<String, Long> map(String word) throws Exception {
            return Tuple2.of(word, 1L);
        }
    }
}


/*

拉姆达表达式:
    java8增加一种函数式编程
    
    如果传入的是一个接口, 并且这个接口只有一个抽象方法, 这种接口叫函数式接口. 可以使用lambda
    
 */